package kafka

import (
	"sync"

	"gitee.com/zawei/yggdrasil/pkg/log"
	"github.com/Shopify/sarama"
)

type OffsetManager interface {
	Read(msg *Message)
	Commit(sess sarama.ConsumerGroupSession, msg *Message)
}

type defaultManager struct {
}

func (m *defaultManager) Read(*Message) {}

func (m *defaultManager) Commit(sess sarama.ConsumerGroupSession, msg *Message) {
	sess.MarkOffset(msg.Topic, msg.Partition, msg.Offset+1, msg.Metadata)
}

type Offset struct {
	Offset int64
	b      bool
}

type sequenceManager struct {
	offsetMu  sync.RWMutex
	offsetSeq map[string]map[int32][]Offset
}

func newSequenceManager() OffsetManager {
	return &sequenceManager{
		offsetSeq: map[string]map[int32][]Offset{},
	}
}

func (m *sequenceManager) Read(msg *Message) {
	m.offsetMu.Lock()
	defer m.offsetMu.Unlock()
	topicMap := m.offsetSeq[msg.Topic]
	partitionSlice, ok := topicMap[msg.Partition]
	if !ok {
		topicMap[msg.Partition] = []Offset{{Offset: msg.Offset}}
		return
	}
	topicMap[msg.Partition] = append(partitionSlice, Offset{Offset: msg.Offset})
}

func (m *sequenceManager) Commit(sess sarama.ConsumerGroupSession, msg *Message) {
	m.offsetMu.Lock()
	defer m.offsetMu.Unlock()
	topicMap := m.offsetSeq[msg.Topic]
	partitionSlice, ok := topicMap[msg.Partition]
	if !ok {
		sess.MarkOffset(msg.Topic, msg.Partition, msg.Offset, msg.Metadata)
		log.Debugf("single offset committed: topic[%s], partition[%s], offset[%d]", msg.Topic, msg.Partition, msg.Offset)
		return
	}
	log.Debugf("offset committed: offset[%d]", msg.Offset)
	partitionSlice[msg.Offset-partitionSlice[0].Offset].b = true
	var maxOffset int64
	i := 0
	for ; i < len(partitionSlice); i++ {
		if partitionSlice[i].b == true {
			maxOffset = partitionSlice[i].Offset
		}
	}
	if maxOffset > 0 {
		sess.MarkOffset(msg.Topic, msg.Partition, maxOffset+1, "")
		log.Debugf("multiple offset committed: topic[%s], partition[%s], offset[%d]", msg.Topic, msg.Partition, msg.Offset)
		if len(partitionSlice) == i {
			delete(topicMap, msg.Partition)
		} else if i > 0 {
			topicMap[msg.Partition] = partitionSlice[i:]
		}
	}
}
